package ch.ethz.inf.shop.model.simpleDatabase.operators;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import ch.ethz.inf.shop.model.simpleDatabase.Tuple;
import ch.ethz.inf.shop.model.simpleDatabase.TupleSchema;

public class GroupBy extends Operator {

	private final List<Tuple> groupList;
	private int cursor = 0;

	public enum AggregateFunction {
		SUM, COUNT
	}

	/**
	 * Example 1:
	 * 
	 * select l.product_id, count(l.product_id) like_cnt from `like` AS l group
	 * by l.product_id; "*" is not allowed
	 * 
	 * Example 2:
	 * 
	 * select recipe.*, sum(product.price) cost_of_ingredients from recipe,
	 * recipe_item, product where recipe.recipe_id=recipe_item.recipe and
	 * product.product_id = recipe_item.product group by recipe.recipe_id;
	 * 
	 * 
	 * 
	 * @param op
	 * @param func
	 *            Aggregate Function SUM or COUNT
	 * @param columnToAggregate
	 *            in Example 1 or price in Example 2
	 * @param aggregatedColumnName
	 *            in Example 1 like_cnt or in Example 2 cost_of_ingredients
	 * @param groupByColumn
	 *            in Example 1 l.product_id in Example 2 recipe.recipe_id
	 * 
	 * 
	 * 
	 */
	public GroupBy(Operator op, AggregateFunction func, String columnToAggregate, String aggregatedColumnName, String groupByColumn) {
		Map<String, Tuple> groups = new HashMap<String, Tuple>();

		// works like this:
		// while there are tuples in the underlying operator
		while (op.moveNext()) {

			// check if there is already a group tuple, for current tuple
			// a group tuple consists of all attributes of the current tuple
			// except the "columnToAggregate" attribute, with an additional
			// attribute "aggregatedColumnName"
			String groupKey = op.current().getString(groupByColumn);
			if (!groups.containsKey(groupKey)) {
				List<String> newColumnNames = new ArrayList<String>(op.current().getSchema().getColumnNames());
				newColumnNames.remove(columnToAggregate);
				newColumnNames.add(aggregatedColumnName);

				TupleSchema schema = new TupleSchema(newColumnNames);
				List<String> newValues = new ArrayList<String>(op.current().getValues());
				newValues.remove(op.current().getSchema().getIndex(columnToAggregate));
				newValues.add("");

				// add the fresh groupTuple to the groups map
				groups.put(groupKey, new Tuple(schema, newValues));

			}

			Tuple groupTuple = groups.get(groupKey);
			// aggregate the current tuple, to its group
			aggregate(groupTuple, aggregatedColumnName, op.current(), columnToAggregate, func);
		}

		groupList = new ArrayList<Tuple>(groups.values());
	}

	private static void aggregate(Tuple groupTuple, String aggregatedColumnName, Tuple current, String columnToAggregate, AggregateFunction func) {
		switch (func) {
		case SUM:
			BigDecimal newVal = groupTuple.getBigDecimal(aggregatedColumnName);

			newVal = newVal.add(current.getBigDecimal(columnToAggregate));
			groupTuple.setBigDecimal(aggregatedColumnName, newVal);
			break;
		case COUNT:
			groupTuple.setInt(aggregatedColumnName, groupTuple.getInt(aggregatedColumnName) + 1);
			break;
		}

	}

	@Override
	public boolean moveNext() {
		if (cursor < groupList.size()) {
			current = groupList.get(cursor++);
			return true;
		}
		return false;
	}

}
